Appearance
Binance Python SPOT SDK
这是一个用于 Binance Spot SDK API 的客户端库,使开发者能够通过编程方式与 Binance 的 SPOT 交易平台进行交互。该库通过以下三个不同的端点提供获取市场数据、执行交易和管理订单的工具:
目录
支持的功能
- REST API 端点:
/api/*
- WebSocket 端点:实时数据流和请求-响应通信。
- 包含测试用例和示例,便于快速上手。
安装
要使用此库,请确保您的环境运行 Python 3.9 或更高版本。
bash
pip install binance-sdk-spot文档
详细信息,请参考 Binance API 文档。
REST API
所有 REST API 端点均可通过 rest_api 模块访问。REST API 可用于获取市场数据、管理交易以及访问账户信息。请注意,部分端点需要使用您的 Binance API 凭证进行身份验证。
python
from binance_common.configuration import ConfigurationRestAPI
from binance_common.constants import SPOT_REST_API_PROD_URL
from binance_sdk_spot.spot import Spot
from binance_sdk_spot.rest_api.models import ExchangeInfoResponse, RateLimits
logging.basicConfig(level=logging.INFO)
configuration = ConfigurationRestAPI(api_key="your-api-key", api_secret="your-api-secret", base_path=SPOT_REST_API_PROD_URL)
client = Spot(config_rest_api=configuration)
try:
response = client.rest_api.exchange_info(symbol="BNBUSDT")
rate_limits: List[RateLimits] = response.rate_limits
logging.info(f"exchange_info() 的速率限制: {rate_limits}")
data: ExchangeInfoResponse = response.data()
logging.info(f"exchange_info() 响应结果: {data}")
except Exception as e:
logging.error(f"exchange_info() 出现错误: {e}")更多示例可在 examples/rest_api 文件夹中找到。
配置选项
REST API 支持以下高级配置选项:
timeout: 请求的超时时间,单位为毫秒(默认值:1000 毫秒)。proxy: 代理配置:host: 代理服务器主机名。port: 代理服务器端口。protocol: 代理协议(http 或 https)。auth: 代理身份验证凭据:username: 代理用户名。password: 代理密码。
keep_alive: 启用 HTTP 长连接(默认值:true)。compression: 启用响应压缩(默认值:true)。retries: 请求失败时的重试次数(默认值:3 次)。backoff: 重试之间的延迟时间,单位为毫秒(默认值:1000 毫秒)。https_agent: 自定义 HTTPS 代理,用于高级 TLS 配置。time_unit: 指定时间戳的时间单位(例如,毫秒或微秒)。private_key: 用于身份验证的 RSA 或 ED25519 私钥。private_key_passphrase: 如果私钥被加密,则为其密码短语。
超时
您可以配置请求的超时时间(以毫秒为单位)。如果请求超过指定的超时时间,则会被中止。有关详细用法,请参见 超时示例。
代理
REST API 支持 HTTP/HTTPS 代理配置。有关详细用法,请参见 代理示例。
长连接
为持久连接启用 HTTP 长连接。有关详细用法,请参见 长连接示例。
压缩
启用或禁用响应压缩。有关详细用法,请参见 压缩示例。
重试
配置请求失败时的重试次数以及每次重试之间的延迟时间(单位为毫秒)。有关详细用法,请参见 重试示例。
HTTPS 代理
自定义 HTTPS 代理以进行高级 TLS 配置。有关详细用法,请参见 HTTPS 代理示例。
基于密钥对的身份验证
REST API 支持基于密钥对的安全通信身份验证。您可以使用 RSA 或 ED25519 密钥对请求进行签名。有关详细用法,请参见 基于密钥对的身份验证示例。
时间单位
REST API 支持不同的时间单位用于时间戳值。有关详细信息,请参见 时间单位示例。
证书锁定
为了增强安全性,您可以在配置中使用 https_agent 选项进行证书锁定。这确保了客户端仅与使用特定证书的服务器通信。有关详细用法,请参见 证书锁定示例。
错误处理
REST API 提供了详细的错误类型,以帮助您有效处理问题:
ClientError: 表示 SDK 客户端中发生的错误。RequiredError: 当缺少或未定义必填参数时抛出。UnauthorizedError: 表示缺少或无效的身份验证凭据。ForbiddenError: 禁止访问所请求的资源。TooManyRequestsError: 超出速率限制。RateLimitBanError: 因超出速率限制而被禁止的 IP 地址。ServerError: 内部服务器错误,可选地包含状态码。NetworkError: 网络连接问题。NotFoundError: 资源未找到。BadRequestError: 无效请求或无法处理的请求。
有关详细用法,请参见 错误处理示例。
测试网
为了测试目的,现货测试网中可以使用 /api/* 端点。请在配置中更新 base_path:
python
from binance_common.configuration import ConfigurationRestAPI
from binance_common.constants import SPOT_REST_API_TESTNET_URL
from binance_sdk_spot.spot import Spot
configuration = ConfigurationRestAPI(api_key="your-api-key", api_secret="your-api-secret", base_path=SPOT_REST_API_TESTNET_URL)
client = Spot(config_rest_api=configuration)如果未提供 base_path,则默认使用 https://api.binance.com。
WebSocket API
WebSocket API 为市场数据和交易操作提供请求-响应通信。使用 websocket_api 模块与这些端点进行交互。
python
import asyncio
import logging
from binance_common.configuration import ConfigurationWebSocketAPI
from binance_common.constants import SPOT_WS_API_PROD_URL
from binance_sdk_spot.spot import Spot
from binance_sdk_spot.websocket_api.models import AccountCommissionResponse
logging.basicConfig(level=logging.INFO)
configuration_ws_api = ConfigurationWebSocketAPI(api_key="your-api-key", api_secret="your-api-secret", stream_url=SPOT_WS_API_PROD_URL)
client = Spot(config_ws_api=configuration_ws_api)
async def exchange_info():
connection = None
try:
connection = await client.websocket_api.create_connection()
response = await connection.account_commission(
symbol="BNBUSDT",
)
result: AccountCommissionResponse = response.data()
print(f"交易所信息: {result}")
except Exception as e:
print(f"exchange_info() 错误: {e}")
finally:
if connection:
await connection.close_connection(close_session=True)
if __name__ == "__main__":
asyncio.run(exchange_info())更多示例可在 examples/websocket_api 文件夹中找到。
配置选项
WebSocket API 支持以下高级配置选项:
timeout:设置 WebSocket 请求的超时时间(默认:5000 毫秒)。reconnect_delay:重连之间的延迟时间(毫秒)。compression:启用响应压缩。proxy:代理配置:host:代理服务器主机名。port:代理服务器端口。protocol:代理协议(http 或 https)。auth:代理身份验证凭据:username:代理用户名。password:代理密码。
mode:选择single或pool连接模式。single:单个 WebSocket 连接。pool:一组 WebSocket 连接。
pool_size:在池模式下定义 WebSocket 连接的数量。time_unit:指定时间戳的时间单位(例如,毫秒或微秒)。private_key:用于身份验证的 RSA 或 ED25519 私钥。private_key_passphrase:私钥的密码短语(如果加密的话)。https_agent:用于高级 TLS 配置的自定义 HTTPS 代理。user_agent:用于 WebSocket 流的自定义用户代理字符串。
超时
以毫秒为单位设置 WebSocket API 请求的超时时间。有关详细用法,请参阅 超时示例。
重连延迟
指定 WebSocket 重连尝试之间的延迟时间(毫秒)。有关详细用法,请参阅 重连延迟示例。
压缩
启用或禁用 WebSocket 消息的压缩。有关详细用法,请参阅 压缩示例。
代理
WebSocket API 支持 HTTP/HTTPS 代理配置。有关详细用法,请参阅 代理示例。
连接模式
选择 WebSocket 连接的 single 或 pool 连接模式。single 模式使用单个 WebSocket 连接,而 pool 模式使用一组 WebSocket 连接。有关详细用法,请参阅 连接模式示例。
时间单位
指定 WebSocket API 时间戳的时间单位(例如,毫秒或微秒)。有关详细用法,请参阅 时间单位示例。
密钥对身份验证
使用 RSA 或 ED25519 私钥进行 WebSocket API 身份验证。有关详细用法,请参阅 密钥对身份验证示例。
WebSocket HTTP 代理
自定义代理以进行高级配置。有关详细用法,请参阅 WebSocket HTTP 代理示例。
测试网络
为了测试目的,WebSocket API 还支持测试网络环境。请在您的配置中更新 wsURL:
python
from binance_common.configuration import ConfigurationWebSocketAPI
from binance_common.constants import SPOT_WS_API_TESTNET_URL
from binance_sdk_spot.spot import Spot
configuration_ws_api = ConfigurationWebSocketAPI(
api_key="your-api-key",
api_secret="your-api-secret",
stream_url=SPOT_WS_API_TESTNET_URL
)
client = Spot(config_ws_api=configuration_ws_api)如果未提供 wsURL,则默认使用 wss://ws-api.binance.com:443/ws-api/v3。
WebSocket 流
WebSocket 流为市场交易、K线图等提供实时数据推送。请使用 websocket-streams 模块订阅这些数据流。
python
import asyncio
import logging
from binance_common.configuration import ConfigurationWebSocketStreams
from binance_common.constants import SPOT_WS_STREAMS_PROD_URL
from binance_sdk_spot.spot import Spot
logging.basicConfig(level=logging.INFO)
configuration_ws_streams = ConfigurationWebSocketStreams(
stream_url=SPOT_WS_STREAMS_PROD_URL,
)
client = Spot(config_ws_streams=configuration_ws_streams)
async def agg_trade():
connection = None
try:
connection = await client.websocket_streams.create_connection()
stream = await connection.agg_trade(
symbol="bnbusdt",
)
stream.on("message", lambda data: print(f"{data}"))
await asyncio.sleep(5)
await stream.unsubscribe()
except Exception as e:
logging.error(f"agg_trade() 错误: {e}")
finally:
if connection:
await connection.close_connection(close_session=True)
if __name__ == "__main__":
asyncio.run(agg_trade())更多示例可在 examples/websocket-streams 文件夹中找到。
配置选项
WebSocket Streams API 支持以下高级配置选项:
reconnect_delay:重连之间的延迟时间(毫秒)。compression:启用响应压缩。proxy:代理配置:host:代理服务器主机名。port:代理服务器端口。protocol:代理协议(http 或 https)。auth:代理身份验证凭据:username:代理用户名。password:代理密码。
mode:选择single或pool连接模式。single:单个 WebSocket 连接。pool:一组 WebSocket 连接。
pool_size:在池模式下定义 WebSocket 连接的数量。time_unit:指定时间戳的时间单位(例如,毫秒或微秒)。https_agent:用于高级 TLS 配置的自定义 HTTPS 代理。user_agent:用于 WebSocket Streams 的自定义用户代理字符串。
重连延迟
为流指定 WebSocket 重连尝试之间的延迟时间(毫秒)。详细用法请参见 重连延迟示例。
压缩
启用或禁用 WebSocket Streams 消息的压缩功能。详细用法请参见 压缩示例。
代理
WebSocket Streams 支持 HTTP/HTTPS 代理配置。详细用法请参见 代理示例。
连接模式
为 WebSocket Streams 选择 single 或 pool 连接模式。single 模式使用单个 WebSocket 连接,而 pool 模式使用一组 WebSocket 连接。详细用法请参见 连接模式示例。
时间单位
为 WebSocket Streams 的时间戳指定时间单位(例如,毫秒或微秒)。详细用法请参见 时间单位示例。
WebSocket HTTP 代理
自定义代理以进行高级配置。详细用法请参见 WebSocket HTTP 代理示例。
取消订阅流
您可以使用 unsubscribe 方法取消订阅特定的 WebSocket 流。这在无需关闭连接的情况下管理活动订阅时非常有用。
python
import asyncio
import logging
from binance_common.configuration import ConfigurationWebSocketStreams
from binance_sdk_spot.spot import Spot
logging.basicConfig(level=logging.INFO)
client = Spot(config_ws_streams=ConfigurationWebSocketStreams())
async def agg_trade():
connection = None
try:
connection = await client.websocket_streams.create_connection()
stream = await connection.agg_trade(
symbol="bnbusdt",
)
stream.on("message", lambda data: print(f"{data}"))
await asyncio.sleep(5)
await stream.unsubscribe()
except Exception as e:
logging.error(f"agg_trade() error: {e}")
finally:
if connection:
await connection.close_connection(close_session=True)
if __name__ == "__main__":
asyncio.run(agg_trade())测试网
Websocket 流也支持用于开发和测试的测试网环境。请在你的配置中更新 wsURL:
python
from binance_common.configuration import ConfigurationWebSocketStreams
from binance_common.constants import SPOT_WS_API_TESTNET_URL
from binance_sdk_spot.spot import Spot
config_ws_streams = ConfigurationWebSocketStreams(
stream_url=SPOT_WS_STREAMS_TESTNET_URL
)
client = Spot(config_ws_streams=config_ws_streams)如果未提供 wsURL,则默认使用 wss://stream.binance.com:9443。
自动连接续期
在 API 密钥 24 小时过期之前,WebSocket 连接(包括 WebSocket API 和 WebSocket 流)会自动续期,以确保持续连接。
测试
要运行测试,请确保你已安装 Poetry,然后执行以下命令:
bash
poetry install
poetry run pytest ./tests测试覆盖范围包括:
- REST API 端点
- WebSocket API 和数据流
- 错误处理和边界情况
迁移指南
如果你正在升级到新的模块化结构,请参考 迁移指南 以获取详细步骤。
贡献
我们欢迎贡献!
由于此仓库包含自动生成的代码,我们建议你首先在 GitHub 上创建一个 issue 来讨论你的想法或提出改进建议。这有助于确保所做的更改符合项目目标和自动生成流程。
要进行贡献,请按以下步骤操作:
- 在 GitHub 上创建一个 issue,描述你的建议或发现的 bug。
- 如果确认需要更改,维护人员将把更改合并到主分支中。
如果你直接进行贡献,请确保所有测试都能通过。请在讨论并确认更改后再提交 pull request。
感谢你的贡献!
许可证
本项目采用 MIT 许可证。详情请参阅 LICENCE 文件。
